package cn.com.cennavi.services;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;

import scala.Tuple2;
import cn.com.cennavi.bean.MineLabODBean;
import cn.com.cennavi.bean.OD_HotSpot_OutDataBean;
import cn.com.cennavi.bean.OD_OdNum_OutDataBean;
import cn.com.cennavi.bean.ShpFileBean;
import cn.com.cennavi.util.MapUtil;

import com.vividsolutions.jts.geom.Geometry;
import com.vividsolutions.jts.geom.GeometryFactory;
import com.vividsolutions.jts.geom.MultiPolygon;
import com.vividsolutions.jts.geom.PrecisionModel;
import com.vividsolutions.jts.io.ParseException;
import com.vividsolutions.jts.io.WKTReader;

/**
 * 采用轨迹数据最小粒度为用户自定义的od数量统计模型
 * 
 * @Description
 * @author Thinkpad
 * @date 2017年2月20日 上午10:00:23
 */
public class ODAnalysisTrajectoryODHotspot implements PairFlatMapFunction<MineLabODBean, String, OD_HotSpot_OutDataBean> {

	/**
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么)
	 */
	private static final long serialVersionUID = 1L;
	private SimpleDateFormat sf = new SimpleDateFormat("yyyyMMddHHmmss");
	private List<ShpFileBean> dataList = new ArrayList<ShpFileBean>();
	private int spatialFlag, meshCode, countyCode, cityCode, province;
	private int timegranularityFlag;
	private float timegranularity;
	private int spatialgranularityFlag;
	private Map<Integer, Set<Integer>> meshCountyMap = new HashMap<Integer, Set<Integer>>();

	public ODAnalysisTrajectoryODHotspot(Broadcast<List<ShpFileBean>> mapBroad, int spatialFlag, int meshCode, int countyCode, int cityCode, int province, int timegranularityFlag, float timegranularity, Broadcast<Map<Integer, Set<Integer>>> meshCountyMapBroad, int spatialgranularityFlag) {
		this.dataList = mapBroad.value();
		this.spatialFlag = spatialFlag;
		this.meshCode = meshCode;
		this.countyCode = countyCode;
		this.cityCode = cityCode;
		this.province = province;
		this.timegranularityFlag = timegranularityFlag;
		this.timegranularity = timegranularity;
		this.spatialgranularityFlag = spatialgranularityFlag;
		this.meshCountyMap = meshCountyMapBroad.value();
		System.out.println(dataList.size());
	}

	@Override
	public ArrayList<Tuple2<String, OD_HotSpot_OutDataBean>> call(MineLabODBean arg0) throws Exception {
		// TODO Auto-generated method stub
		ArrayList<Tuple2<String, OD_HotSpot_OutDataBean>> returnList = new ArrayList<Tuple2<String, OD_HotSpot_OutDataBean>>();
		MineLabODBean tb = arg0;
		long otime = tb.getOTime();
		long dtime = tb.getDTime();
		Date odate = new Date((long) otime * 1000);
		Date ddate = new Date((long) dtime * 1000);
		String odateStr = sf.format(odate);
		String ddateStr = sf.format(ddate);
		double olat = tb.getOLatitude();
		double olon = tb.getOLongitude();
		double dlat = tb.getDLatitude();
		double dlon = tb.getDLongitude();
//		String o_minGridId = MapUtil.findCell(olon, olat, 2500);
//		String d_minGridId = MapUtil.findCell(dlon, dlat, 2500);
		int o_meshId = MapUtil.findMesh(olon, olat);
		int d_meshId = MapUtil.findMesh(dlon, dlat);
		Set<Integer> o_county = this.meshCountyMap.get(o_meshId) == null ? null : this.meshCountyMap.get(o_meshId);
		Set<Integer> d_county = this.meshCountyMap.get(d_meshId) == null ? null : this.meshCountyMap.get(d_meshId);
		if (o_county == null || d_county == null) {
			returnList.add(new Tuple2<String, OD_HotSpot_OutDataBean>("errorData:" + arg0.toDataString(), null));
			return returnList;
		}
		int o_city = cityCode;
		int d_city = cityCode;
		int o_province = province;
		int d_province = province;
		String o_bestGridId = "null";
		String d_bestGridId = "null";
		switch (spatialgranularityFlag) {
		// 0代表二次网格,1代表自定义网格
		case 0:
			o_bestGridId = MapUtil.findMesh(olon, olat) + "";
			d_bestGridId = MapUtil.findMesh(dlon, dlat) + "";
			break;
		case 1:
			o_bestGridId = getbestGridId(olon, olat, dataList) + "";
			d_bestGridId = getbestGridId(dlon, dlat, dataList) + "";
			if (o_bestGridId.equals("-1") || d_bestGridId.equals("-1")) {
				returnList.add(new Tuple2<String, OD_HotSpot_OutDataBean>("errorData:" + arg0.toDataString(), null));
				return returnList;
			}
			break;
		default:
			break;
		}
		int oTimeId = -1, dTimeId = -1;
		switch (timegranularityFlag) {
		case 0:
			// 时间粒度单位是分钟， 获取od点的时间粒度id,最小粒度30，则一天被分成48份
			oTimeId = getMinFlag(odateStr, timegranularity);
			dTimeId = getMinFlag(ddateStr, timegranularity);
			break;
		default:
			// 扩展
			break;
		}
		for (Integer ocounty : o_county) {
			for (Integer dcounty : d_county) {
				OD_HotSpot_OutDataBean oBean = new OD_HotSpot_OutDataBean();
				OD_HotSpot_OutDataBean dBean = new OD_HotSpot_OutDataBean();
				oBean.date = Integer.parseInt(odateStr.substring(0, 8));
				oBean.weekday = getWeekDay(odate);
				oBean.timeGranularity = (int) timegranularity;
				oBean.timeIndex = oTimeId;
				//目前minGridId为二次网格id
//				oBean.minGridId = Integer.parseInt(o_minGridId);
				oBean.minGridId = Integer.parseInt(o_bestGridId);
				oBean.bestGridId = Integer.parseInt(o_bestGridId);
				oBean.county = ocounty;
				oBean.city = o_city;
				oBean.province = o_province;
				oBean.oNum = 1;

				dBean.date = Integer.parseInt(ddateStr.substring(0, 8));
				dBean.weekday = getWeekDay(ddate);
				dBean.timeGranularity = (int) timegranularity;
				dBean.timeIndex = dTimeId;
//				dBean.minGridId = Integer.parseInt(d_minGridId);
				dBean.minGridId = Integer.parseInt(d_bestGridId);
				dBean.bestGridId = Integer.parseInt(d_bestGridId);
				dBean.county = dcounty;
				dBean.city = d_city;
				dBean.province = d_province;
				dBean.dNum = 1;
				if (countyIsIncity(oBean))
					returnList.add(new Tuple2<String, OD_HotSpot_OutDataBean>(oBean.toDataString(), oBean));
				if (countyIsIncity(dBean))
					returnList.add(new Tuple2<String, OD_HotSpot_OutDataBean>(dBean.toDataString(), dBean));
			}
		}

		return returnList;
	}

	/**
	 * 当按照区域代码、城市编码、省份时选数据时，数据的gps点需要先计算mesh，然后找county，mesh和county是一对多的，
	 * 如果其中一个county是待选county
	 * ，则认为当前点满足要求，但是在生成数据的时候会把其他county也生成，这样就会有一些冗余数据，需要过滤掉
	 * 
	 * @Description
	 * @author buf
	 * @param oBean
	 * @param spatialFlag
	 * @return
	 */
	private boolean countyIsIncity(OD_HotSpot_OutDataBean oBean) {
		// TODO Auto-generated method stub
		System.out.println("niuxinzan:" + spatialFlag + "," + oBean.county + "," + countyCode);
		switch (spatialFlag) {
		case 0:
			return oBean.county == countyCode ? true : false;
		case 1:
			if ((oBean.county / 10000) == 11 || (oBean.county / 10000) == 31 || (oBean.county / 10000) == 12 || (oBean.county / 10000) == 50) {
				return (oBean.county / 10000) == (oBean.city / 10000) ? true : false;
			} else {
				return (oBean.county / 100) == (oBean.city / 100) ? true : false;
			}

		case 2:
			return (oBean.county/ 10000) == (oBean.city / 10000) ? true : false;
		
		}
		return false;
	}

	Calendar calendar = Calendar.getInstance();

	private int getWeekDay(Date d) {
		// TODO Auto-generated method stub
		calendar.setTime(d);
		return calendar.get(Calendar.DAY_OF_WEEK);
	}

	// private String gatcounty(Set<Integer> o_county) {
	// // TODO Auto-generated method stub
	// return o_county.iterator().next() + "";
	// }

	private int getbestGridId(double lon, double lat, List<ShpFileBean> dataList) throws ParseException {
		// TODO Auto-generated method stub
		for (ShpFileBean sbs : dataList) {
			MultiPolygon geo = sbs.shpPoint;
			int srid = 4326;
			GeometryFactory factory = new GeometryFactory(new PrecisionModel(), srid);
			WKTReader reader = new WKTReader(factory);
			Geometry geom = reader.read("POINT (" + lon + " " + lat + ")");
			if (geo.covers(geom)) {
				return sbs.shpId;
			}
		}
		return -1;
	}

	/**
	 * 获取分钟下标
	 * 
	 * @Description
	 * @author buf
	 * @param odateStr
	 * @param timegranularity
	 * @return
	 */
	private int getMinFlag(String odateStr, float timegranularity) {
		// TODO Auto-generated method stub
		// TODO Auto-generated method stub
		// float unit = (float) 24 / (float) timegranularity;
		int timeSplits = (int) (1440 / timegranularity);
		int hour = Integer.parseInt(odateStr.substring(8, 10));
		int min = Integer.parseInt(odateStr.substring(10, 12));
		int mimutes = hour * 60 + min;
		int index = (int) (mimutes / timegranularity);
		return index;
	}

	/**
	 * 获取小时坐标
	 * 
	 * @Description
	 * @author buf
	 * @param odateStr
	 * @param timegranularity
	 * @return
	 */
	private int getHourFlag(String odateStr, float timegranularity) {
		// TODO Auto-generated method stub
		// TODO Auto-generated method stub
		// float unit = (float) 24 / (float) timegranularity;
		int hour = Integer.parseInt(odateStr.substring(8, 10));
		int min = Integer.parseInt(odateStr.substring(10, 12));
		int hour1 = (int) (hour / timegranularity);
		int hour2 = 0;
		if (timegranularity < 1)
			hour2 = (int) (min / (60 * timegranularity));
		return hour1 + hour2;
	}

}
